package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.TrafficPageViewBean;
import com.atguigu.realtime.bean.UserRegisterBean;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Dws_UserUserAddActWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_UserUserAddActWindow().init(
                4003,
                2,
                "Dws_UserUserAddActWindow",
                "dwd_user_register"

        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
//        SingleOutputStreamOperator<JSONObject> pageString = filterPage(stream);
//        AllWindowedStream<JSONObject, TimeWindow> jsonObjectTimeWindowAllWindowedStream = windowA(pageString);

        stream
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                                .withTimestampAssigner((obj, ts) -> obj.getLong("ts"))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject obj, long recordTimestamp) {
                                        //TODO 把创建时间作为水印
                                        Long ts = AtguiguUtil.toShootTime(obj.getString("create_time"));
//                                        System.out.println(ts);
                                        return ts;
                                    }
                                })
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(
                        new AggregateFunction<JSONObject, Long, UserRegisterBean>() {
                            //创建累加器
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            //累加的方法
                            @Override
                            public Long add(JSONObject value, Long acc) {
                                acc += 1;
                                return acc;
                            }

                            @Override
                            public UserRegisterBean getResult(Long acc) {
                                return new UserRegisterBean("", "", acc, System.currentTimeMillis());
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return a+b;
                            }
                        },
                        new ProcessAllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx, Iterable<UserRegisterBean> iterable, Collector<UserRegisterBean> out) throws Exception {
                                UserRegisterBean bean = iterable.iterator().next();

                                bean.setStt(AtguiguUtil.toDatTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDatTime(ctx.window().getEnd()));

                                out.collect(bean);
                            }
                        }
                )
                .addSink(FlinkSinkUtil.getClickHouseSink("dws_user_user_register_window", UserRegisterBean.class)
                );
    }

//    private SingleOutputStreamOperator<UserRegisterBean> windowA(SingleOutputStreamOperator<UserRegisterBean> pageString) {
//        return pageString.assignTimestampsAndWatermarks(
//                WatermarkStrategy
//                        .<UserRegisterBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                        .withTimestampAssigner((bean, ts) -> bean.getTs())
//        )
//        .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .reduce(
//                        new ReduceFunction<UserRegisterBean>() {
//                            @Override
//                            public UserRegisterBean reduce(UserRegisterBean value1, UserRegisterBean value2) throws Exception {
//                                value1.setRegisterCt(value1.getRegisterCt() + value2.getRegisterCt());
//                                return value1;
//                            }
//                        },
//                        new ProcessAllWindowFunction<UserRegisterBean, UserRegisterBean, TimeWindow>() {
//                            @Override
//                            public void process(Context ctx, Iterable<UserRegisterBean> iterable, Collector<UserRegisterBean> out) throws Exception {
//                                UserRegisterBean bean = iterable.iterator().next();
//
//                                bean.setStt(AtguiguUtil.toDatTime(ctx.window().getStart()));
//                                bean.setEdt(AtguiguUtil.toDatTime(ctx.window().getEnd()));
//
//                                bean.setTs(System.currentTimeMillis());
//
//                                out.collect(bean);
//                            }
//                        }
//                );
//
//    }
//
//    private SingleOutputStreamOperator<JSONObject> filterPage(DataStreamSource<String> stream) {
//        return stream.map(JSON::parseObject)
//                .map(new RichMapFunction<JSONObject, TrafficPageViewBean>() {
//
//            private ValueState<String> dateState;
//
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                dateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("dateState", String.class));
//            }
//
//            @Override
//            public TrafficPageViewBean map(JSONObject obj) throws Exception {
//                JSONObject common = obj.getJSONObject("common");
//                JSONObject page = obj.getJSONObject("page");
//
//                String vc = common.getString("vc");
//                String ch = common.getString("ch");
//                String ar = common.getString("ar");
//                String isNew = common.getString("is_new");
//                Long ts = obj.getLong("ts");
//
//                Long pvCt = 1L;
//                Long durSum = page.getLong("during_time");
//
//                Long svCt = 0L;
//                String lastPageId = page.getString("last_page_id");
//                if (lastPageId == null || lastPageId.length() == 0) {
//                    svCt = 1L;
//                }
//
//                Long uvCt = 0L;
//                // 状态中存储年月日, 如果来的数据的日期和状态不一样, 就是当天第一条 uvCt=1 更新状态,
//                // 否则就不是第一条
//                String date = dateState.value();
//                String today = AtguiguUtil.toDate(ts);
//                if (!today.equals(date)) {
//                    // 更新状态
//                    dateState.update(today);
//                    uvCt = 1L;
//                }
//
//                return new TrafficPageViewBean("", "",
//                        vc, ch, ar, isNew,
//                        uvCt, svCt, pvCt, durSum,
//                        ts);
//            }
//        };
//    }

}
